package com.css.wc;

import com.esotericsoftware.kryo.io.Output;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;

import java.util.Arrays;
import java.util.Map;

/**
 * 输入一行数据
 * 计算
 * 输出，单词及单词数显的次数
 */
public class SplitBolt extends BaseRichBolt {
    private OutputCollector collector;

    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        this.collector = outputCollector;
    }

    @Override
    public void execute(Tuple tuple) {
        //获取上游句子
        String juzi = tuple.getStringByField("juzi");
        //切割
        String[] split = juzi.split(" ");
        //发送数据
        //单词及单词数显的次数
        for(String word:split){
            collector.emit(Arrays.asList(word,"1"));
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("word","number"));
    }
}
